package vn.ebuzz.sms.worker;

import java.io.BufferedReader;
import java.io.InputStream;
import java.io.InputStreamReader;
import java.io.UnsupportedEncodingException;
import java.sql.Connection;
import java.sql.PreparedStatement;
import java.sql.ResultSet;
import java.sql.SQLException;
import java.sql.Statement;
import java.util.ArrayList;
import java.util.List;

import org.apache.http.HttpResponse;
import org.apache.http.NameValuePair;
import org.apache.http.client.HttpClient;
import org.apache.http.client.entity.UrlEncodedFormEntity;
import org.apache.http.client.methods.HttpPost;
import org.apache.http.impl.client.DefaultHttpClient;
import org.apache.http.message.BasicNameValuePair;
import org.apache.http.params.BasicHttpParams;
import org.apache.http.params.HttpConnectionParams;
import org.apache.http.params.HttpParams;
import org.apache.log4j.Logger;

import vn.ebuzz.sms.core.AESGenerator;
import vn.ebuzz.sms.core.Configuration;
import vn.ebuzz.sms.core.Database;
import vn.ebuzz.sms.core.Outgoing;

public class OutgoingQueue {
	private static Logger log = Logger.getLogger("ouqueue");
	private static OutgoingQueue q = null;
	volatile boolean isRunning = true;

	private OutgoingQueue(){
		for(int i = 0; i < Configuration.MAX_THREAD; i ++){
			OutgoingWorker t = new OutgoingWorker(i);
			t.start();
			log.debug("Start thread " + t.getId());	
			// sleep between thread
			try {
				Thread.sleep(Configuration.SEPERATOR);
			} catch (InterruptedException e) {
				e.printStackTrace();
			}			
		}
	}

	public static OutgoingQueue getInstance(){
		if(q == null)
			q = new OutgoingQueue();
		return q;
	}

	public void stop(){
		isRunning = false;
	}

	private class OutgoingWorker extends Thread{
		private int index;
		public OutgoingWorker(int i){
			this.index = i;
		}
		@Override
		public void run() {
			long threadId = Thread.currentThread().getId();			
			while(isRunning){
				long start = System.currentTimeMillis();
				log.debug("OutgoingWorker" + threadId + " check DB");
				ArrayList<Outgoing> myList = new ArrayList<Outgoing>();

				Connection conn = Database.getConnection();
				Statement st = null;
				ResultSet rs = null;
				try{
					st = conn.createStatement();
					rs = st.executeQuery("select id, sender, receiver," +
							" message, mo_id, session_id," +
							" unique_id, type, url from outgoing" +
							" where status = " + Configuration.MT_QUEUED +
							" and mod(id, " + Configuration.MAX_THREAD + ") = " + index +
					" order by created asc LIMIT " + Configuration.MAX_BATCH);
					while(rs.next()){
						Outgoing out = new Outgoing();
						out.setId(rs.getInt("id"));
						out.setSender(rs.getString("sender"));
						out.setReceiver(rs.getString("receiver"));
						out.setMessage(rs.getString("message"));
						out.setMo_id(rs.getString("mo_id"));
						out.setSession_id(rs.getString("session_id"));
						out.setUnique_id(rs.getString("unique_id"));
						out.setMttype(rs.getInt("type"));
						out.setUrl(rs.getString("url"));
						myList.add(out);
					}
				} catch(Exception e){
					log.error("Error", e);
				} finally{
					if(rs != null)
						try {
							rs.close();
						} catch (SQLException e) {
							log.error("Error", e);
						}
						if(st != null)
							try {
								st.close();
							} catch (SQLException e) {
								log.error("Error", e);
							}
							Database.closeConnection(conn);					
				}
				log.debug("OutgoingWorker" + threadId + " get " + myList.size() + " jobs");
				if(myList.size() > 0){
					// TODO: call httpclient to send MT
					for(int i = 0; i < myList.size(); i ++){
						Outgoing out = myList.get(i);
						String result = sendMessage(out);
						if(result.trim().equalsIgnoreCase("<h5>0</h5>")){
							// success
							out.setStatus(Configuration.MT_SENT);
							out.setResult(result.trim());
						}
						else{
							// failed
							out.setStatus(Configuration.MT_FAILED);
							out.setResult(result);
						}
					}
					// TODO: save result to DB, batch update here
					UpdateRecordsIntoOutgoing(myList);
					long end = System.currentTimeMillis();
					long exectime = end - start;
					if(exectime < Configuration.THRESHOLD){
						try {
							Thread.sleep(exectime);
						} catch (InterruptedException e) {
							log.error("Error", e);
						}						
					}
				}
				else{
					try {
						Thread.sleep(Configuration.SLEEP);
					} catch (InterruptedException e) {
						log.error("Error", e);
					}
				}
				log.debug("OutgoingWorker" + threadId + " finish " + myList.size() + " jobs");
			}
			log.debug("OutgoingWorker" + threadId + " stop");
		}

		private String sendMessage(Outgoing out){
			//TODO: send message via HTTP POST
			log.debug("Start send MT to :" + Configuration.VIVASURL);
			StringBuilder sb = new StringBuilder();

			HttpParams httpParams = new BasicHttpParams();
			HttpConnectionParams.setSoTimeout(httpParams, 10000);
			HttpConnectionParams.setConnectionTimeout(httpParams, 10000);
			HttpClient httpClient = new DefaultHttpClient(httpParams);

			HttpPost httpPost = new HttpPost(Configuration.VIVASURL);
			List<NameValuePair> nameValuePairs = new ArrayList<NameValuePair>(2);
			nameValuePairs.add(new BasicNameValuePair("cpId", Configuration.CPID));
			nameValuePairs.add(new BasicNameValuePair("data", encryptData(out, Configuration.CPKEY)));
			nameValuePairs.add(new BasicNameValuePair("uniqueId", out.getUnique_id()));

			try {
				httpPost.setEntity(new UrlEncodedFormEntity(nameValuePairs));
			} catch (UnsupportedEncodingException e) {
				log.error("Error", e);
			}
			try {
				HttpResponse response = httpClient.execute(httpPost);
				InputStream is = response.getEntity().getContent();
				BufferedReader reader = new BufferedReader(
						new InputStreamReader(is));

				String line;
				while ((line = reader.readLine()) != null) {
					sb.append(line);
				}
				log.info("Send MT done with code " + sb.toString());
				httpClient.getConnectionManager().closeExpiredConnections();
				httpClient.getConnectionManager().shutdown();

				return sb.toString();
			} catch (Exception e) {
				log.error("Error", e);
				return "exception : send MT failed";
			}
		}
		
		private String encryptData(Outgoing out, String cpkey){
			String encrypt = "sId=" + out.getSession_id();
			encrypt += "|moId=" + out.getMo_id();
			encrypt += "|msisdn=" + out.getReceiver();
			encrypt += "|sn=" + out.getSender();
			encrypt += "|time=" + System.currentTimeMillis();
			encrypt += "|mtEnd=1";
			encrypt += "|mtType=" + out.getMttype();
			encrypt += "|msg=" + out.getMessage();
			encrypt += "|mtUrl=" + out.getUrl(); 
			log.debug("Send MT " + encrypt);
			encrypt = AESGenerator.encrypt(encrypt, cpkey);
			return encrypt;
		}
		
		private  void UpdateRecordsIntoOutgoing(List<Outgoing> listOutgoings){
			log.debug("Insert batch incoming " + listOutgoings.size());

			Connection conn = null;
			PreparedStatement preparedStatement = null;

			String insertTableSQL = "update outgoing set status = ? ,result= ?  where mo_id = ?";

			try {
				conn = Database.getConnection();
				preparedStatement = conn.prepareStatement(insertTableSQL);

				conn.setAutoCommit(false);
				for(Outgoing outgoing : listOutgoings){
					preparedStatement.setInt(1, outgoing.getStatus());
					preparedStatement.setString(2, outgoing.getResult());
					preparedStatement.setString(3, outgoing.getMo_id());
					preparedStatement.addBatch();
				}
				preparedStatement.executeBatch();
				conn.commit();
				conn.setAutoCommit(true);
			} catch (SQLException e) {
				log.error("Error", e);
				try {
					conn.rollback();
				} catch (SQLException e1) {
					log.error("Error", e);
				}
			} finally {
				if (preparedStatement != null) {
					try {
						preparedStatement.close();
					} catch (SQLException e) {
						log.error("Error", e);
					}
				}
				if (conn != null) {
					Database.closeConnection(conn);
				}

			}		
		}
	}
}
